package com.shujia.window



import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
import org.apache.flink.util.Collector

object Demo3Window {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    val students: DataStream[String] = env.readTextFile("data/students.txt")
    //val students: DataStream[String] = env.socketTextStream("master",8888)

    /**
      * 统计最近每个班级10个学生的平均年龄，每隔一个学生统计一次
      */

    val kvDS: DataStream[(String, Int)] = students.map(student =>{
      val split: Array[String] = student.split(",")
      (split(4),split(2).toInt)
    })

    kvDS
      .keyBy(_._1)
      .countWindow(10,1)
        .process(new MyProcessWindowFunction).print()


    env.execute()


  }

}


class MyProcessWindowFunction extends ProcessWindowFunction[(String,Int),(String,Double),String,GlobalWindow]{

  /** process每一个窗口中的每一个key运行一次
    *
    * @param key 班级
    * @param context  上下对象
    * @param elements  班级在这个窗口中的所有数据
    * @param out  将数据发送到下游
    */
  override def process(key: String, context: Context, elements: Iterable[(String, Int)], out: Collector[(String, Double)]): Unit = {

    //取出年龄
    val ages: List[Int] = elements.toList.map(_._2)
    //计算平均年龄
    val avgAge: Double = ages.sum.toDouble / ages.length

    //将数据发送到下游
    out.collect((key,avgAge))

  }

}
